package com.hex.ds.hdtp.core.app.data.table.service.impl;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.hex.ds.hdtp.core.app.common.enums.BasicListStatus;
import com.hex.ds.hdtp.core.app.common.page.PageData;
import com.hex.ds.hdtp.core.app.common.page.converter.IPageAndPageDtoConverter;
import com.hex.ds.hdtp.core.app.common.page.dto.PageDto;
import com.hex.ds.hdtp.core.app.data.table.converter.ITableTransferListConverter;
import com.hex.ds.hdtp.core.app.data.table.converter.ITableTransferTaskConverter;
import com.hex.ds.hdtp.core.app.data.table.dto.request.TableTransferListRequestDto;
import com.hex.ds.hdtp.core.app.data.table.dto.request.TableTransferRunLogRequestDto;
import com.hex.ds.hdtp.core.app.data.table.dto.request.TableTransferTaskRequestDto;
import com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferListResponseDto;
import com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferTaskResponseDto;
import com.hex.ds.hdtp.core.app.data.table.service.ITableTransferListService;
import com.hex.ds.hdtp.core.app.data.table.service.ITableTransferRunLogService;
import com.hex.ds.hdtp.core.app.data.table.service.ITableTransferTaskService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceConfInfoService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceInfoService;
import com.hex.ds.hdtp.core.inf.common.page.Page;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcConnectUtil;
import com.hex.ds.hdtp.core.inf.data.table.po.TableTransferTaskPo;
import com.hex.ds.hdtp.core.inf.data.table.repository.ITableTransferTaskRepository;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceInfoPo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;

/**
 * @Package com.hex.ds.hdtp.core.app.data.table.service.impl
 * @ClassName TableTransferTaskService
 * @Description 表迁移任务-业务接口实现
 * @Author Wang zhihao
 * @Date 2023/8/8 18:40
 * @Version v1.0
 **/
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class TableTransferTaskService implements ITableTransferTaskService {

    private final ITableTransferTaskRepository tableTransferTaskRepository;

    private final ITableTransferTaskConverter converter;

    private final ITableTransferListConverter iTableTransferListConverter;

    private final IPageAndPageDtoConverter pageAndPageDtoConverter;

    @Lazy
    private final ITableTransferListService tableTransferListService;

    private final ISourceInfoService sourceInfoService;

    private final ISourceConfInfoService sourceConfInfoService;

    private final ITableTransferRunLogService tableTransferRunLogService;

    /**
     * @param tableTransferTaskRequestDto
     * @param pageDto
     * @Method queryListByPage
     * @Param
     * @Return com.hex.ds.hdtp.core.app.common.page.PageData
     * @Description 表迁移任务-分页查询
     * @Author Wang zhihao
     * @Date 2023/8/8 19:25
     * @Version V1.0
     */
    @Override
    public PageData queryListByPage(TableTransferTaskRequestDto tableTransferTaskRequestDto, PageDto pageDto) {
        final Page<TableTransferTaskPo> pageResult = tableTransferTaskRepository.queryListByPage(
                converter.toTableTransferTaskPo(tableTransferTaskRequestDto), pageAndPageDtoConverter.toPage(pageDto));
        List<TableTransferTaskResponseDto> responseDtos = converter.toTableTransferTaskResponseDtoList(pageResult.getData());
        return new PageData(responseDtos, pageResult.getTotal());
    }

    /**
     * @param tableTransferTaskRequestDto
     * @Method add
     * @Param
     * @Return void
     * @Description 表迁移任务-新增
     * @Author Wang zhihao
     * @Date 2023/8/9 10:53
     * @Version V1.0
     */
    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void add(TableTransferTaskRequestDto tableTransferTaskRequestDto) {
        try {
            tableTransferTaskRequestDto.setPkId(null);
            tableTransferTaskRepository.addWithBackPkId(converter.toTableTransferTaskPo(tableTransferTaskRequestDto));
        } catch (Exception e) {
            throw new RuntimeException("新增表迁移任务失败" + e.getMessage());
        }
    }

    /**
     * @param tableTransferTaskRequestDto
     * @Method queryList
     * @Param
     * @Return java.util.List<com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferTaskResponseDto>
     * @Description 表迁移任务-不分页查询
     * @Author Wang zhihao
     * @Date 2023/8/10 18:25
     * @Version V1.0
     */
    @Override
    public List<TableTransferTaskResponseDto> queryList(TableTransferTaskRequestDto tableTransferTaskRequestDto) {
        List<TableTransferTaskPo> tableTransferTask =
                tableTransferTaskRepository.queryList(converter.toTableTransferTaskPo(tableTransferTaskRequestDto));
        return converter.toTableTransferTaskResponseDtoList(tableTransferTask);
    }

    /**
     * @param tableTransferTaskRequestDto
     * @Method addWithBackId
     * @Param
     * @Return java.lang.String
     * @Description 新增任务并返回PkId
     * @Author Wang zhihao
     * @Date 2023/8/10 15:28
     * @Version V1.0
     */
    @Override
    public String addWithBackId(TableTransferTaskRequestDto tableTransferTaskRequestDto) {
        tableTransferTaskRequestDto.setTransferTotal(0);
        tableTransferTaskRequestDto.setTransferSuccessCount(0);
        tableTransferTaskRequestDto.setTransferFailCount(0);
        return tableTransferTaskRepository.addWithBackPkId(converter.toTableTransferTaskPo(tableTransferTaskRequestDto));
    }


    /**
     * @param tableTransferTaskRequestDto
     * @Method modifyById
     * @Param
     * @Return void
     * @Description 表迁移-根据主键修改
     * @Author Wang zhihao
     * @Date 2023/8/9 10:51
     * @Version V1.0
     */
    @Override
    public void modifyById(TableTransferTaskRequestDto tableTransferTaskRequestDto) {
        int count = tableTransferTaskRepository.modifyById(converter.toTableTransferTaskPo(tableTransferTaskRequestDto));
        if (count != 1) {
            throw new RuntimeException("修改表迁移任务失败");
        }
    }

    /**
     * @param pkId
     * @Method removeById
     * @Param
     * @Return void
     * @Description 表迁移任务-根据pkId删除
     * @Author Wang zhihao
     * @Date 2023/8/9 10:46
     * @Version V1.0
     */
    @Override
    public void removeById(String pkId) {
        int count = tableTransferTaskRepository.removeById(pkId);
        tableTransferListService.removeByTaskPkId(pkId);
        if (count != 1) {
            throw new RuntimeException("删除表迁移任务失败");
        }
    }

    /**
     * @param pkIds
     * @Method batchRemoveById
     * @Param
     * @Return void
     * @Description 表迁移任务-批量删除
     * @Author Wang zhihao
     * @Date 2023/8/9 10:44
     * @Version V1.0
     */
    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void batchRemoveById(List<String> pkIds) {
        for (String pkId : pkIds) {
            removeById(pkId);
        }
    }

    /**
     * @param pkId
     * @Method transfer
     * @Param
     * @Return void
     * @Description 表迁移
     * @Author Wang zhihao
     * @Date 2023/8/10 14:20
     * @Version V1.0
     */
    @Override
    public void transfer(String pkId) {
        // 初始化信息
        tableTransferTaskRepository.modifyById(converter.toTableTransferTaskPo(new TableTransferTaskRequestDto()
                .setPkId(pkId).setStartTime(DateUtil.format(new DateTime(), "yyyy-MM-dd HH:mm:ss"))
                .setTransferSuccessCount(0).setTransferFailCount(0)));
        // 根据表迁移任务获取目标信息
        SourceInfoPo targetConnInfoPo = sourceInfoService.queryByPkId(queryInfoById(pkId).getTargetConnPkId());
        List<SourceConfInfoPo> sourceConfInfoPoList = sourceConfInfoService.queryListBySourceCode(targetConnInfoPo.getSourceCode());
        // 遍历该任务下的建表清单并执行建表
        List<TableTransferListResponseDto> tableTransferLists = queryTransferListByTaskPkId(pkId);
        for (TableTransferListResponseDto tableTransfer : tableTransferLists) {
            String executeSql = tableTransfer.getTargetCreateSql();
            Boolean executeSqlFlag = Boolean.FALSE;
            String message = "表迁移成功";
            try {
                executeSqlFlag = JdbcConnectUtil.runSql(sourceConfInfoPoList, executeSql);
            } catch (Exception e) {
                message = e.getMessage();
            }
            modifyTableTransferRunLog(pkId, tableTransfer, executeSqlFlag, message);
        }
        tableTransferTaskRepository.modifyById(converter.toTableTransferTaskPo(
                new TableTransferTaskRequestDto().setPkId(pkId).setEndTime(DateUtil.format(new DateTime(), "yyyy-MM-dd HH:mm:ss"))));
    }

    /**
     * @param pkId
     * @param tableTransfer
     * @param createFlag
     * @Method modifyTableTransferRunLog
     * @Param
     * @Return void
     * @Description 更新建表情况
     * @Author Wang zhihao
     * @Date 2023/8/16 11:34
     * @Version V1.0
     */
    private void modifyTableTransferRunLog(String pkId, TableTransferListResponseDto tableTransfer, Boolean createFlag, String message) {
        if (createFlag.equals(Boolean.TRUE)) {
            updateListStatus(tableTransfer, BasicListStatus.SUCCESS.name());
            tableTransferTaskRepository.modifySuccessNum(pkId);
            tableTransferTaskRepository.modifyTransferStatus(DateUtil.now(), pkId);
        } else {
            updateListStatus(tableTransfer, BasicListStatus.FAIL.name());
            tableTransferTaskRepository.modifyFailNum(pkId);
            tableTransferTaskRepository.modifyTransferStatus(DateUtil.now(), pkId);
        }
        addRunLog(tableTransfer, message);
    }

    /**
     * @param tableTransfer
     * @Method updateListStatus
     * @Param
     * @Return void
     * @Description 更新建表清单的建表状态
     * @Author Wang zhihao
     * @Date 2023/8/16 10:58
     * @Version V1.0
     */
    private void updateListStatus(TableTransferListResponseDto tableTransfer, String status) {
        TableTransferListRequestDto tableTransferListRequestDto = iTableTransferListConverter.toTableTransferListRequestDto(tableTransfer);
        tableTransferListRequestDto.setStatus(status);
        tableTransferListRequestDto.setPartitionFieldNameList(Arrays.asList(tableTransfer.getPartitionFieldName().toUpperCase().split(",")));
        tableTransferListService.modifyById(tableTransferListRequestDto);
    }

    /**
     * @param tableTransfer
     * @Method addRunLog
     * @Param
     * @Return void
     * @Description 添加建表记录
     * @Author Wang zhihao
     * @Date 2023/8/13 22:53
     * @Version V1.0
     */
    private void addRunLog(TableTransferListResponseDto tableTransfer, String message) {
        TableTransferRunLogRequestDto tableTransferRunLogRequestDto = iTableTransferListConverter.toTableTransferRunLog(tableTransfer, message);
        tableTransferRunLogService.add(tableTransferRunLogRequestDto);
    }

    /**
     * @param pkId
     * @Method queryInfoById
     * @Param
     * @Return com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferTaskResponseDto
     * @Description 根据主键获取建表任务信息
     * @Author Wang zhihao
     * @Date 2023/8/13 22:11
     * @Version V1.0
     */
    @Override
    public TableTransferTaskResponseDto queryInfoById(String pkId) {
        return converter.toTableTransferTaskResponseDto(tableTransferTaskRepository.queryInfoById(pkId));
    }

    /**
     * @param pkId
     * @param type
     * @Method modifyTransferTotal
     * @Param
     * @Return void
     * @Description 根据主键更改表迁移清单总数
     * @Author Wang zhihao
     * @Date 2023/8/25 9:50
     * @Version V1.0
     */
    @Override
    public void modifyTransferTotal(String pkId, String type) {
        switch (type) {
            case "increase":
                tableTransferTaskRepository.increaseTotalNum(pkId);
                break;
            case "decrease":
                tableTransferTaskRepository.decreaseTotalNum(pkId);
                break;
            default:
                break;
        }
    }

    /**
     * @param pkId
     * @Method queryTransferListByTaskPkId
     * @Param
     * @Return void
     * @Description 根据任务主键查询表迁移清单
     * @Author Wang zhihao
     * @Date 2023/8/10 14:23
     * @Version V1.0
     */
    private List<TableTransferListResponseDto> queryTransferListByTaskPkId(String pkId) {
        TableTransferListRequestDto tableTransferListRequestDto = new TableTransferListRequestDto().setTaskPkId(pkId);
        return tableTransferListService.queryList(tableTransferListRequestDto);
    }
}
